package com.clown.orderPayDetect

import java.util

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.cep.{PatternSelectFunction, PatternTimeoutFunction}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object OrderTimeoutWithoutCep {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val resource = getClass.getResource("/OrderLog.csv")
    val inputStream = env.readTextFile(resource.getPath)

    val orderEventStream = inputStream
      .map(var2 => {
        val arr = var2.split(",")
        OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    // 自定义ProcessFunction进行复杂事件的检测
    val orderTimeoutOutputTag = new OutputTag[OrderResult]("orderTimeout")
    val orderResultStream = orderEventStream
      .keyBy(_.orderId)
      .process(new OrderPayMatchResult(orderTimeoutOutputTag))

    orderResultStream.print("payed")
    orderResultStream.getSideOutput(orderTimeoutOutputTag).print("timeout")
    env.execute("order timeout without cep  job")
  }
}

// 自定义实现KeyedProcessFunction
class OrderPayMatchResult(tag: OutputTag[OrderResult]) extends KeyedProcessFunction[Long, OrderEvent, OrderResult] {
  // 定义状态
  lazy val isCreatedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("isCreatedState", classOf[Boolean]))
  lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("isPayedState", classOf[Boolean]))
  lazy val timestampState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timestampState", classOf[Long]))

  override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, out: Collector[OrderResult]): Unit = {
    // 先拿到当前状态
    val isPayed = isPayedState.value()
    val isCreated = isCreatedState.value()
    val timestamp = timestampState.value()
    // 判断当前事件类型，看是create还是pay
    // 1. 来的是create,要继续判断是否pay过
    if (value.eventType.equals("create")) {
      // 1.1 如果已经支付过，正常支付，输出匹配成功的结果
      if (isPayed) {
        out.collect(OrderResult(value.orderId, "payed successfully"))
        isCreatedState.clear()
        isPayedState.clear()
        timestampState.clear()
        ctx.timerService().deleteEventTimeTimer(timestamp)
      } else {
        // 1.2 如果还没pay过，注册定时器，等待15分钟
        val ts = value.timestamp * 1000L + 900 * 100L
        ctx.timerService().registerEventTimeTimer(ts)
        timestampState.update(ts)
        isCreatedState.update(true)
      }
    }
    // 2. 如果当前来的是pay，要判断是否create过
    else if (value.eventType.equals("pay")) {
      if (isCreated) {
        // 2.1 如果已经create过，匹配成功，还要判断一下pay时间是否超过了定时器时间
        if (value.timestamp * 1000 < timestamp) {
          // 2.1.1 没有超时，正常输出
          out.collect(OrderResult(value.orderId, "payed successfully"))
        } else {
          // 2.1.2 已经超时，输出超时结果
          ctx.output(tag, OrderResult(value.orderId, "payed but already timeout"))
        }
        // 只要输出结果，当前order处理已经结束，清空状态和定时器
        isCreatedState.clear()
        isPayedState.clear()
        timestampState.clear()
        ctx.timerService().deleteEventTimeTimer(timestamp)
      } else {
        // 2.2 如果create没来，注册定时器，等到pay的时间就可以
        ctx.timerService().registerEventTimeTimer(value.timestamp * 1000L)
        // 更新状态
        timestampState.update(value.timestamp * 1000L)
        isPayedState.update(true)
      }
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
    // 定时器触发
    // 1. pay来了，没等到create
    if (isPayedState.value()) {
      ctx.output(tag, OrderResult(ctx.getCurrentKey, "payed but not found create log"))
    } else {
      // 2. create来了，没有pay
      ctx.output(tag, OrderResult(ctx.getCurrentKey, "order timeout"))
    }
    // 清空状态
    isCreatedState.clear()
    isPayedState.clear()
    timestampState.clear()
  }
}